package com.atuigu.userprofile.app;

import com.atguigu.bean.TagInfo;
import com.atguigu.constans.ConstCodes;
import com.atguigu.dao.TagInfoDAO;
import com.atguigu.util.MyClickhouseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class TaskBitmapApp {
    public static void main(String[] args) {
        //将宽表中的数据根据不同的标签值的类型写入到不同的bitmap表

        //因为代码最终是部署在spark环境下运行的，因此代码中需要初始化sparkContext   SparkSession包含SparkContext
        SparkConf sparkConf = new SparkConf().setAppName("TaskBitmapApp");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        //获取外部传参
        String busiDate = args[1];

        //获取启用的标签
        List<TagInfo> tagInfoList = TagInfoDAO.getTagInfoWithOn();

        //根据不同的标签指的类型创建四个集合
        ArrayList<TagInfo> longTagInfolist = new ArrayList<>();
        ArrayList<TagInfo> decimalTagInfolist = new ArrayList<>();
        ArrayList<TagInfo> stringTagInfolist = new ArrayList<>();
        ArrayList<TagInfo> dateTagInfolist = new ArrayList<>();

        //根据标签值类型，将遍历出来的所有已启用标签的数据放入不同的集合中
        for (TagInfo info : tagInfoList) {
            if (ConstCodes.TAG_VALUE_TYPE_LONG.equals(info.getTagValueType())) {
                longTagInfolist.add(info);
            } else if (ConstCodes.TAG_VALUE_TYPE_DECIMAL.equals(info.getTagValueType())) {
                decimalTagInfolist.add(info);
            } else if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(info.getTagValueType())) {
                stringTagInfolist.add(info);
            } else if (ConstCodes.TAG_VALUE_TYPE_DATE.equals(info.getTagValueType())) {
                dateTagInfolist.add(info);
            }
        }

        //针对不同的集合调用方法
        insertBitmap(longTagInfolist, busiDate, "user_tag_value_long");
        insertBitmap(decimalTagInfolist, busiDate, "user_tag_value_decimal");
        insertBitmap(stringTagInfolist, busiDate, "user_tag_value_string");
        insertBitmap(dateTagInfolist, busiDate, "user_tag_value_date");


    }

    public static void insertBitmap(List<TagInfo> list, String busiDate, String bitmapTableName) {

        if (list.size()>0) {


            List<String> tagValueWithTagCodeList = list.stream().map(tagInfo -> "(" + tagInfo.getTagCode().toLowerCase() + ",'" + tagInfo.getTagCode().toLowerCase() + "')").collect(Collectors.toList());

            String tagValueWithTagCodeSQL = StringUtils.join(tagValueWithTagCodeList, ",");

//        System.out.println(tagValueWithTagCodeSQL);

            //获取款表的表名
            String tableName = "up_tag_merge_" + busiDate.replace("-", "");

            //编写查询语句
            /**
             * select groupBitmapState(uid),tp.1 tag_value,tp.2 tag_code from (
             *                                               select uid,
             *                                                      arrayJoin(
             *                                                              [(gender, 'gender'), (agegroup, 'agegroup'), (favor, 'favor')]) as tp
             *                                               from user_tag_merge
             * ) group by tag_value,tag_code;
             */
            String querySQL = "  select " +
                    "tp.2 tag_code," +
                    "tp.1 tag_value," +
                    "groupBitmapState(cast(uid as UInt64))," +
                    "'"+busiDate+"' "+
                    " from (\n" +
                    "                                                        select uid,\n" +
                    "                                                               arrayJoin(\n" +
                    "                                                                       [" + tagValueWithTagCodeSQL + "]) as tp\n" +
                    "                                                        from " + tableName + "\n" +
                    "          ) group by tag_value,tag_code";


            //保证幂等性  在数据出现问题重跑的时候，将脏数据删除掉
            String deleteSQL = "alter table " + bitmapTableName + " delete where dt = '" + busiDate + "'";


            //编写插入语句  插入的表根据标签值类型来确定
            String insertSQL = "insert into " + bitmapTableName + querySQL;
            System.out.println(insertSQL);

            //执行删除语句
            MyClickhouseUtil.executeSql(deleteSQL);

            //执行插入语句
            MyClickhouseUtil.executeSql(insertSQL);
        }
    }
}

//TODO 我现在要把所以已启用的标签 根据不同的标签值类型 分为四类并放一起 每一类有   配套的表  插入语句（表名）